//package com.kafka.streams;
//
//import org.apache.kafka.common.serialization.Serdes;
//import org.apache.kafka.common.utils.Bytes;
//import org.apache.kafka.streams.KeyValue;
//import org.apache.kafka.streams.StreamsBuilder;
//import org.apache.kafka.streams.StreamsConfig;
//import org.apache.kafka.streams.kstream.KStream;
//import org.apache.kafka.streams.kstream.Materialized;
//import org.apache.kafka.streams.kstream.TimeWindows;
//import org.apache.kafka.streams.processor.WallclockTimestampExtractor;
//import org.apache.kafka.streams.state.WindowStore;
//import org.springframework.context.annotation.Bean;
//import org.springframework.kafka.annotation.EnableKafkaStreams;
//import org.springframework.kafka.config.KafkaStreamsConfiguration;
//
//import java.util.HashMap;
//import java.util.Map;
//import java.util.concurrent.TimeUnit;
//
///**
// * @author zhangbowen
// * @date 2019/4/2
// **/
//@EnableKafkaStreams
//public class KafkaStreams {
//    @Bean
//    public KafkaStreamsConfiguration defaultKafkaStreamsConfig() {
//        Map<String, Object> props = new HashMap<>();
//        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "testStreams1");
//        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
//        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.Integer().getClass().getName());
//        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
//        props.put(StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, WallclockTimestampExtractor.class.getName());
//        return new KafkaStreamsConfiguration(props);
//    }
//
//    /**
//     * kafkaStreams总结(kStream->kStream、KStream<->KTable、KTable<->KTable支持连接;含table字段都是非窗口连接)
//     * (KGroupedStream->KTable支持先窗口移动或者不移动再聚合、KGroupedTable->KTable窗口不能移动只能直接聚合 )
//     *
//     * @param kStreamBuilder
//     * @return
//     */
//    @Bean
//    public KStream<Integer, String> kStream(StreamsBuilder kStreamBuilder) {
//        KStream<Integer, String> stream = kStreamBuilder.stream("streams1");
//        long segmentTime = TimeUnit.MINUTES.toMillis(5);
//        stream.groupByKey()
//                .windowedBy(TimeWindows.of(segmentTime).advanceBy(segmentTime))
//                //.reduce((oldValue,newValue) -> oldValue)
//                .aggregate(() -> "", (key, newValue, oldValue) -> oldValue,
//                        Materialized.<Integer, String, WindowStore<Bytes, byte[]>>as("sss")
//                                .withKeySerde(Serdes.Integer())
//                                .withValueSerde(Serdes.String()))
//                .toStream()
//                .map((windowedId, value) -> new KeyValue<>(windowedId.key(), value))
//                .to("topic3");
//        return stream;
//    }
//}
